package day09.producer_and_consumer


import java.util.HashMap

import org.apache.kafka.clients.producer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

object producerDemo {
  /**
    * 模拟kafka的生产者，生产消息后，在consumerDemo这个模拟消费者里获取数据
    * ：(1)用Array(brokers,topic) =  args封装传递过来的信息zk的brokers信息 和  topic名
    *   (2)定义zk的连接的配置消息，即定义HashMap用k,v去存储配置信息。
    *     固定的写法：
    *       .put(producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,  zk信息 )
    *       .put(producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    *       .put(producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    *   (3)创建kafka生产者对象：new KafkaProducer[String, String](将配置变量放进来)
    *   (4)数据。
    *   (5) 生产者.send(new ProducerRecord("往哪个topic名里发送","要发送的消息"))
    *
    *   运行的时候，先将zk/topic名这两个参数输入到args里然后开启生产者；在将zk/group/topics/numThread这四个参数输入到消费者args里然后开启消费者。
    * @param args
    */

  def main(args: Array[String]): Unit = {
    // metadataBrokerList：kafka列表，topic：topic名称，
    // messagesPerSec：每秒的消息数，wordsPerMessage：每秒的单词数量
    if (args.length < 2) {
      System.err.println("Usage: KafkaWordCountProducer <metadataBrokerList> <topic>")
      // 退出程序
      // system.exit(0):正常退出，相当于shell的kill
      // system.exit(1):非正常退出，相当于shell的kill -9
      System.exit(1)
    }
    // args: master:9092,slave1:9092,slave2:9092 mykafkawc
    val Array(brokers, topic) = args
    // zk连接的配置信息
    val props = new HashMap[String, Object]()
    // 注意ProducerConfig是org.apache.kafka.clients.producer这个包的
    props.put(producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")

    // 定义生产者
    val my_producer = new KafkaProducer[String, String](props)

    //数据
    val arr = Array(
      "hello tom",
      "hello fuhao",
      "hello shuaige",
      "hello xiaohua"
    )

    val r = new Random()
    // 发送消息
    while (true){
      val message = arr (r.nextInt(arr.length))
      my_producer.send(new ProducerRecord[String,String ]("mykafkawc", message))
      Thread.sleep(1000)
    }


  }
}
